package com.zhss.microservice.server.slot;

import com.alibaba.fastjson.JSONObject;
import com.zhss.microservice.server.constant.MessageType;
import com.zhss.microservice.server.node.network.ServerMessageReceiver;
import com.zhss.microservice.server.node.network.ServerNetworkManager;
import com.zhss.microservice.server.slot.registry.ServiceInstance;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.zhss.microservice.server.node.persist.FilePersistUtils;

import java.nio.ByteBuffer;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 槽位数据管理组件
 */
public class SlotManager {

    private static final Logger LOGGER = LoggerFactory.getLogger(SlotManager.class);

    /**
     * 槽位分配存储文件的名字
     */
    public static final String NODE_SLOTS_FILENAME = "node_slots";
    public static final String NODE_SLOTS_REPLICAS_FILENAME = "node_slots_replicas";
    public static final Integer SLOT_COUNT = 16384;

    private SlotManager() {

    }

    static class Singleton {
        static SlotManager instance = new SlotManager();
    }

    public static SlotManager getInstance() {
        return Singleton.instance;
    }

    /**
     * 本节点负责管理的槽位集合
     */
    private Slots slots = new Slots();
    /**
     * 本节点负责管理的槽位副本集合
     */
    private Map<String, SlotsReplica> slotsReplicas = new ConcurrentHashMap<>();

    /**
     * 初始化本节点负责的槽位集合
     */
    public void initSlots(List<String> slotsList) {
        if(slotsList == null) {
            ServerMessageReceiver serverMessageReceiver = ServerMessageReceiver.getInstance();
            slotsList = serverMessageReceiver.takeNodeSlots();
        }

        for(String slotScope : slotsList) {
            slots.init(slotScope);
        }
        FilePersistUtils.persist(JSONObject.toJSONString(slotsList).getBytes(), NODE_SLOTS_FILENAME);

        LOGGER.info("初始化本节点槽位数据完毕......");
    }

    /**
     * 初始化本节点负责的槽位副本的集合
     */
    public void initSlotsReplicas(List<String> slotScopes, boolean isController) {
        ServerMessageReceiver serverMessageReceiver = ServerMessageReceiver.getInstance();
        if(slotScopes == null && !isController) {
            slotScopes = serverMessageReceiver.takeNodeSlotsReplicas();
        } else if(slotScopes == null && isController) {
            return;
        }

        for(String slotScope : slotScopes) {
            SlotsReplica slotsReplica = new SlotsReplica();
            slotsReplica.init(slotScope);
            slotsReplicas.put(slotScope, slotsReplica);
        }

        byte[] bytes = JSONObject.toJSONString(slotScopes).getBytes();
        FilePersistUtils.persist(bytes, NODE_SLOTS_REPLICAS_FILENAME);
        LOGGER.info("初始化本节点槽位副本数据完毕......");
    }

    /**
     * 初始化该节点负责的槽位集合的副本节点id
     */
    public void initReplicaNodeId(Integer replicaNodeId) {
        ServerMessageReceiver serverMessageReceiver = ServerMessageReceiver.getInstance();
        if(replicaNodeId == null) {
            replicaNodeId = serverMessageReceiver.takeReplicaNodeId();
        }
        slots.setReplicaNodeId(replicaNodeId);
        LOGGER.info("初始化副本节点id完毕：" + replicaNodeId);
    }

    public void refreshReplicaNodeId(Integer newReplicaNodeId) {
        LOGGER.info("副本节点id进行刷新，老节点id：" + slots.getReplicaNodeId() + "，新节点id：" + newReplicaNodeId);
        slots.setReplicaNodeId(newReplicaNodeId);
    }

    /**
     * 获取槽位
     * @param serviceName
     * @return
     */
    public Slot getSlot(String serviceName) {
        return slots.getSlot(routeSlot(serviceName));
    }

    /**
     * 获取槽位副本
     * @param serviceName
     * @return
     */
    public Slot getSlotReplica(String serviceName) {
        Integer slotNo = routeSlot(serviceName);

        SlotsReplica  slotsReplica = null;

        for(String slotScope : slotsReplicas.keySet()) {
            Integer startSlot = Integer.valueOf(slotScope.split(",")[0]);
            Integer endSlot = Integer.valueOf(slotScope.split(",")[1]);

            if(slotNo >= startSlot && slotNo <= endSlot) {
                slotsReplica = slotsReplicas.get(slotScope);
                break;
            }
        }

        return slotsReplica.getSlot(slotNo);
    }

    /**
     * 将副本转正为slots
     * @param replicaSlotsList
     */
    public void changeReplicaToSlots(List<String> replicaSlotsList) {
        // 把这些slots从当前节点管理的副本槽位里拿出来
        // 把拿出来的副本slots都转移到正式的槽位数据集合里去
        for(String replicaSlots : replicaSlotsList) {
            SlotsReplica slotsReplica = slotsReplicas.get(replicaSlots);
            ConcurrentHashMap<Integer, Slot> _slots = slotsReplica.getSlots();
            for(Integer slotNo : _slots.keySet()) {
                Slot slot = _slots.get(slotNo);
                slots.putSlot(slotNo, slot);
            }
        }

        for(String replicaSlots : replicaSlotsList) {
            slotsReplicas.remove(replicaSlots);
        }

        LOGGER.info("副本槽位（" + replicaSlotsList + "）全部进行转正......");
    }

    /**
     * 刷新槽位副本
     * @param replicaSlotsList
     */
    public void refreshReplicaSlots(List<String> replicaSlotsList) {
        LOGGER.info("槽位副本数据进行刷新，老槽位副本包含：" + slotsReplicas.keySet() + "，新槽位副本数据为：" + replicaSlotsList);

        for(String replicaSlots : replicaSlotsList) {
            if(!slotsReplicas.containsKey(replicaSlots)) {
                SlotsReplica slotsReplica = new SlotsReplica();
                slotsReplica.init(replicaSlots);
                slotsReplicas.put(replicaSlots, slotsReplica);
            }
        }

        byte[] bytes = JSONObject.toJSONString(replicaSlotsList).getBytes();
        FilePersistUtils.persist(bytes, NODE_SLOTS_REPLICAS_FILENAME);
    }

    /**
     * 进行槽位数据的转移
     * @param targetNodeId
     * @param slots
     */
    public void transferSlots(Integer targetNodeId, String slots) {
        String[] slotsSplited = slots.split(",");
        Integer startSlotNo = Integer.valueOf(slotsSplited[0]);
        Integer endSlotNo = Integer.valueOf(slotsSplited[1]);

        for(int slotNo = startSlotNo ; slotNo <= endSlotNo; slotNo++) {
            Slot slot = this.slots.getSlot(slotNo);
            // 判断他是否有数据
            if (slot.isEmpty()) {
                continue;
            }
            // 如果他里面有数据的话，就把他里面的数据提取出来，发送给目标节点就可以了
            // 目标节点接收到数据之后，就放到自己的内存的slot里去
            byte[] bytes = slot.getSlotData();

            ByteBuffer buffer = ByteBuffer.allocate(4 + 4 + bytes.length);
            buffer.putInt(MessageType.UPDATE_SLOTS);
            buffer.putInt(slotNo);
            buffer.put(bytes);

            ServerNetworkManager serverNetworkManager = ServerNetworkManager.getInstance();
            serverNetworkManager.sendMessage(targetNodeId, buffer);

            // 再把这个槽位范围从这里都删除掉
            this.slots.removeSlot(slotNo);
        }
    }

    /**
     * 将服务路由到slot
     * @param serviceName
     * @return
     */
    private Integer routeSlot(String serviceName) {
        int hashCode = serviceName.hashCode() & Integer.MAX_VALUE;
        Integer slot = hashCode % SLOT_COUNT;

        if(slot == 0) {
            slot = slot + 1;
        }

        return slot;
    }

    public Integer getReplicaNodeId() {
        return slots.getReplicaNodeId();
    }

    public void updateSlotData(Integer slotNo, List<ServiceInstance> serviceInstances) {
        Slot slot = this.slots.getSlot(slotNo);
        slot.updateSlotData(serviceInstances);
    }

}
